package com.diors.commons.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;

import com.diors.commons.log.DiorsLogger;

/**
 * 选择器线程池执行类
 * @author darren.ouyang
 * @date 2013-9-14
 */
public class SelecterThreadPoolExecutor 
{

	private final DiorsLogger log = DiorsLogger.getLogger(getClass()); 
	
	/* 默认线程数 */
	private final static int defultPoolSize = 3;
	
	/* 减少线程最小任务数 */
	private final static int reduceThreshold = 3;
	
	/* 增加线程最小任务数 */
	private final static int increaseThreshold = 50;
	
	/* 最大线程数 */
	private volatile int maximumPoolSize;
	
	/* 最小线程数 */
	private volatile int minimumPoolSize;
	
	/* 当前线程数 */
	private volatile int poolSize;
	
	/* 初始化线程数 */
	private volatile int initPoolSize;
	
	/* 线程工厂 */
	private volatile ThreadFactory threadFactory;
	
	/* 线程列表 */
	private final List<Worker> workList = new ArrayList<Worker>();
	
	public SelecterThreadPoolExecutor (int initPoolSize, ThreadFactory threadFactory, String checkThreadName, int maxPoolSize, int minPoolSize)
	{
		this.poolSize = 0;
		this.maximumPoolSize = maxPoolSize;
		this.minimumPoolSize = minPoolSize;
		this.initPoolSize = initPoolSize;
		this.threadFactory = threadFactory;
		
		for (int i = 0; i < this.initPoolSize; i++)
			addWoker(newBlockingQueue());
		
		new Thread(new CheckerTask(), checkThreadName).start();
	}
	
	public SelecterThreadPoolExecutor (int initPoolSize, ThreadFactory threadFactory, String checkThreadName)
	{
		this(initPoolSize, threadFactory, checkThreadName, 10, 2);
	}
	
	public SelecterThreadPoolExecutor ()
	{
		this(defultPoolSize, Executors.defaultThreadFactory(), "Defult-Checker");
	}
	
	public SelecterThreadPoolExecutor (PriorityThreadFactory threadFactory)
	{
		this(defultPoolSize, threadFactory, new StringBuilder(threadFactory.getName())
			.append("-Checker").toString());
	}
	
	private BlockingQueue<Runnable> newBlockingQueue ()
	{
		return new LinkedBlockingQueue<Runnable>();
	}
	
	/**
	 * 添加执行一个任务
	 * @param task	任务实体
	 * @create 2013-9-14
	 */
	public void execute(SelecterWrapper task)
	{
		int index = task.selecterCode() % poolSize;
		Worker worker = workList.get(index);
		worker.addQueue(task);
	}
	
	/**
	 * 获取任务数量
	 * @return
	 * @create 2013-9-14
	 */
	public int workSize ()
	{
		int size = 0;
		for (Worker worker : workList)
			size += worker.workSize();
		
		return size;
	}
	
	/**
	 * 添加一个线程
	 * @param workQueue	工作队列
	 * @return
	 * @create 2013-9-14
	 */
	private synchronized boolean addWoker (BlockingQueue<Runnable> workQueue)
	{
		if (poolSize >= maximumPoolSize)
			return Boolean.FALSE;
		
		Worker work = new Worker(workQueue);
		workList.add(work);
		work.thread.start();
		++poolSize;
		
		return Boolean.TRUE;
	}
	
	/**
	 * 删除一个线程
	 * @create 2013-9-14
	 */
	private synchronized boolean removeWorker ()
	{
		if (poolSize <= minimumPoolSize)
			return Boolean.FALSE;
		
		--poolSize;
		Worker worker = workList.remove(workList.size() - 1);
		worker.disable(true);
		
		return Boolean.TRUE;
	}

	/**
	 * 线程任务实体
	 * @author darren.ouyang
	 * @date 2013-9-14
	 */
	private final class Worker implements Runnable
	{
		/* 执行任务的阻塞队列 */
		private final BlockingQueue<Runnable> workQueue;
		
		/* 线程实体 */
		private final Thread thread;
		
		/* 是否失效 */
		private boolean isDisable;
		
		Worker (BlockingQueue<Runnable> workQueue)
		{
			this.workQueue = workQueue;
			this.thread = threadFactory.newThread(this);
			isDisable = false;
		}
		
		/**
		 * 添加一个任务到队列
		 * @param command	任务实体
		 * @return
		 * @create 2013-9-14
		 */
		private boolean addQueue (Runnable command)
		{
			return workQueue.offer(command);
		}

		/**
		 * 获取当前线程的任务数
		 * @return
		 * @create 2013-9-14
		 */
		private int workSize ()
		{
			return workQueue.size();
		}
		
		/**
		 * 设置线程失效状态
		 * @param isDisable	是否失效
		 * @create 2013-9-14
		 */
		private void disable (boolean isDisable)
		{
			this.isDisable = isDisable;
			checkkillWorker();
		}
		
		/**
		 * 检验杀死线程
		 * @return
		 * @create	2013-9-16	darren.ouyang
		 */
		private boolean checkkillWorker ()
		{
			if (isDisable && workSize() <= 0)
			{
				thread.interrupt();
				log.info("执行选择线程池线程中断!");
				return true;
			}
			return false;
		}
		
		@Override
		public void run() 
		{
			while(true) 
			{
				/* 如果已经失效并且队列中没有课执行的任务了,则摧毁线程 */
				if (checkkillWorker())
					return;
					
				try 
				{
					Runnable runnable = workQueue.take();
					runnable.run();
				} 
				catch (InterruptedException e)
				{
					log.warn("work线程在抽取任务时是巧妙的InterruptedException线程终端了");
				}
				catch (Exception e) 
				{
					log.error("执行选择线程池任务错误" + this.toString(), e);
				}
			}
		}
		
		@Override
		public String toString() 
		{
			String str = new StringBuilder("SelecterThreadPoolExecutor - ").
			append(workList.indexOf(this)).append(" works[").
			append(workSize()).append("]").toString();
			
			return str;
		}
	}
	
	/**
	 * 检测线程任务
	 * @author darren.ouyang
	 * @date 2013-9-14
	 */
	private final class CheckerTask implements Runnable
	{
		/**
		 * 任务检测的间隔时间
		 */
		private final int sleepTime = 15 * 1000;
		
		/**
		 * 最后一次检测的数据包数量
		 */
		private int lastSize = 0;

		@Override
		public void run()
		{
			for(;;)
			{
				try
				{
					Thread.sleep(sleepTime);
				}
				catch (Exception e){}
				
				/*等待执行的数据包数*/
				int sizeNow = workSize();

				if(sizeNow < lastSize && sizeNow < reduceThreshold)
				{
					removeWorker();
				}
				else if(sizeNow > lastSize && sizeNow > increaseThreshold)
				{
					if(!addWoker(newBlockingQueue()) && sizeNow > increaseThreshold * 3)
					{
						log.warn(new StringBuilder("检测到有[").append(sizeNow).
							append("]个包在等待执行。").toString());
					}
				}
				
				int last = lastSize;
				lastSize = sizeNow;
				log.info(new StringBuilder("选择器监控线程池[").append(Thread.currentThread().getName()).
						append("]当前线程数[").append(poolSize).append("]本次任务数[").append(sizeNow).
						append("]上次任务数[").append(last).append("]").toString());
			}
		}
	}
}
